package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.behavior.dto.UpdateArticleMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

/**
 * @author: itheima
 * @create: 2022-02-09 14:44
 */
@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {


    /**
     * 实时统计最近一段时间（10S）内某一篇文章所有操作类型（阅读，点赞，评论，收藏）增量值
     *
     * @param input 监听到话题中数据流，形式如下： key为空  val如下：
     *              {"add":1,"articleId":1483287532010598401,"type":1}
     *              {"add":1,"articleId":1483287532010598401,"type":0}
     *              {"add":1,"articleId":1483287532010598401,"type":2}
     * @return 将文章统计结果输出到目标话题中
     */
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //1.处理数据--从数据流，重新设置数据Key（文章ID）跟  Val（文章信息）
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 处理数据逻辑
             * @param key 原始数据流key
             * @param value 原始数据里val JSON结构 {"add":1,"articleId":1483287532010598401,"type":2}
             * @return
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                System.out.println("处理原始数据 val:" + value);
                return new KeyValue<>(updateArticleMessage.getArticleId().toString(), value);
            }
        });

        //2.设置分组：根据文章ID（数据Key）分组
        KGroupedStream<String, String> groupBy = map.groupByKey();

        //3.设置时间窗口：统计10秒内数据流中当前文章所以行为操作
        TimeWindowedKStream<String, String> windowedKStream = groupBy.windowedBy(TimeWindows.of(Duration.ofSeconds(30)));

        //4.设置聚合：当前文章10秒内，所有行为操作对应的增量值
        //4.1 设置初始化聚合信息 没有聚合结果信息
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                //第一条消息进入后，聚合结果为空
                return null;
            }
        };

        //4.2 对10秒内当前这篇文章 多次操作行为进行聚合
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            /**
             * Aggregator泛型  泛型一：数据流中Key的类型  泛型二：数据流中Val的类型  泛型三：上次聚合后结果 数据类型
             * 聚合后回调函数
             * @param key 数据流中Key的值：文章ID
             * @param value 数据流val的值：文章操作行为
             * @param aggregate 当前文章上次聚合结果（10秒内历史数据汇总）
             * @return 当前文章聚合后多种类型操作增量 结果（JSON字符串）
             */
            @Override
            public String apply(String key, String value, String aggregate) {
                //4.2.1 声明聚合结果对象ArticleStreamMessage
                ArticleStreamMessage result = null;

                //4.2.2 判断上次聚合结果是否有值
                if (StringUtils.isBlank(aggregate)) {
                    //上次聚合结果为空
                    result = new ArticleStreamMessage();
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setCollect(0);
                    result.setLike(0);
                    result.setComment(0);
                    System.out.println("当前文章 " + key + "当前文章10秒内没有聚合结果");
                } else {
                    //说明上次聚合结果有值：{articleID:1000,view:1,like:1,comment:1,collect:1}
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                    System.out.println("当前文章：" + key + "，上次聚合结果：" + result);
                }
                //4.2.3 根据当前本次操作行为增量值，更新文章聚合结果对象
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //本次操作增量值
                Integer add = updateArticleMessage.getAdd();
                //跟上次聚合结果进行累加
                switch (updateArticleMessage.getType()) {
                    case 0:
                        //阅读
                        result.setView(result.getView() + add);
                        break;
                    case 1:
                        //点赞
                        result.setLike(result.getLike() + add);
                        break;
                    case 2:
                        //评论
                        result.setComment(result.getComment() + add);
                        break;
                    case 3:
                        //收藏
                        result.setCollect(result.getCollect() + add);
                        break;
                }
                System.out.println("本文章聚合结果："+result);
                return JSON.toJSONString(result);
            }
        };

        //4.根据key值进行聚合
        KTable<Windowed<String>, String> aggregate = windowedKStream.aggregate(init, aggregator);


        //5.将实时计算结果：当前文章对应所有操作的增量值，发送到目标话题中
        KStream<String, String> kStream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return kStream;
    }

}
